package com.huan.gateway.flow;


import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

public class FlowTest {

    /**
     * 测试发布者和订阅者
     */
    @Test
    public void SubmissionPublisherTest() throws InterruptedException {
        //1. 定义一个发布者，发布的数据类型是 Integer
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher();
        //2.定义一个订阅者，来消费发布者的数据

        MySubscriber mySubscriber = new MySubscriber();

        //3. 发布者 和订阅者 建立订阅绑定关系
        publisher.subscribe(mySubscriber);

        int num = 100;
        while (num-- >= 0) {

//            TimeUnit.SECONDS.sleep(2);
            //4. 生产数据，并发布
            publisher.submit(num);

        }

        //5. 当前 main 线程 先等待下，不然不好演示

        TimeUnit.SECONDS.sleep(10);

        //6. 结束后，关闭发布者，在生产环境，这步骤要放到 finally 里面去

        publisher.close();
    }

    /**
     * 测试完整的发布者和订阅者
     */
    @Test
    public void SubmissionPublisherProcessorTest() throws InterruptedException {
        //1. 定义一个发布者，发布的数据类型是 Integer
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher();

        //2. 定义一个数据处理器，它会对数据进行过滤
        MyProcessor myProcessor = new MyProcessor();

        //3 发布者 和 处理者 建立订阅绑定关系，而不是发布者 和订阅者 建立订阅绑定关系了

        publisher.subscribe(myProcessor);

        //4 .定义一个订阅者，来消费发布者的数据
        MySubscriber mySubscriber = new MySubscriber();

        //5. 数据处理者 和订阅者 建立订阅绑定关系
        myProcessor.subscribe(mySubscriber);


        int num = 100;
        while (num-- >= 0) {

//            TimeUnit.SECONDS.sleep(2);
            //4. 生产数据，并发布
            publisher.submit(num);

        }
        //5. 当前 main 线程 先等待下，不然不好演示
        TimeUnit.SECONDS.sleep(10);
        //6. 结束后，关闭发布者，在生产环境，这步骤要放到 finally
        //发布者关闭
        publisher.close();
        //处理器关闭
        myProcessor.close();

    }


    @Test
    public void TestFlux(){
        List<String> list = new ArrayList<>();
        list.add("A");
        list.add("B");
        list.add("C");
        Flux.fromIterable(list)

                //2， 处理 handlerMappings ，最终会返回一个 Mono 异步序列对象
                .concatMap(new Function<String, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(String s) {

                        return Mono.just(s+1);
                    }
                }).next()

                //3. 从 Mono 异步序列对象取出一个，注意不管 Mono 里面有多少个，这步骤之后取出一个
                .subscribe(System.out::println);

    }

}
